/**
 * Copyright 2014 wasu.com
 *
 *
 * Create on 2015年9月21日 下午3:57:42
 */
package xj.toolkit.ebus;

import java.util.HashMap;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadPoolExecutor;

import org.apache.commons.lang3.ArrayUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import xj.toolkit.util.ClosureEx;
import xj.toolkit.util.Functor;

/**
 * TODO：事件放入之前可以判断是否有响应者，如果没有则直接抛弃。
 * 
 * @author <a href="mailto:caoxiaojian@wasu.cn">xiaojiancao</a>
 *
 */
public class AysncSoftBus implements EventBus {

	private static final Logger logger = LoggerFactory
			.getLogger(SimpleBus.class);

	private ThreadPoolExecutor mainExecutor = null;

	/**
	 * 最大的请求队列深度，超过这个数，直接抛出，不接受任何请求。
	 */
	private int maxEventQueueDepth = 2000;

	public int getMaxEventQueueDepth() {
		return maxEventQueueDepth;
	}

	private static class ClosureExSet {
		private Map<UUID, ClosureEx> closures = new HashMap<UUID, ClosureEx>();

		public void add(UUID uuid, ClosureEx closure) {
			closures.put(uuid, closure);
		}

		public void remove(UUID uuid) {
			closures.remove(uuid);
		}

		public void execute(Object... args) {
			for (ClosureEx closure : closures.values()) {
				closure.execute(args);
			}
		}

		public String toString() {
			StringBuilder sb = new StringBuilder();
			for (ClosureEx closure : closures.values()) {
				sb.append(closure.toString());
				sb.append(";");
			}

			return sb.toString();
		}
	}

	private Map<String, ClosureExSet> closureSet = new HashMap<String, ClosureExSet>();

	public AysncSoftBus(int eventActorThreads, int maxEventQueueDepth) {
		mainExecutor = (ThreadPoolExecutor) Executors
				.newFixedThreadPool(eventActorThreads);
		this.maxEventQueueDepth = maxEventQueueDepth;
	}

	private ClosureExSet getOrCreateClosureExSet(String event) {
		ClosureExSet set = closureSet.get(event);

		if (null == set) {
			set = new ClosureExSet();
			closureSet.put(event, set);
		}

		return set;
	}

	private ClosureExSet getClosureExSet(String event) {
		return closureSet.get(event);
	}

	private void doRegisterObserver(final String event, final UUID id,
			final ClosureEx closure) {
		getOrCreateClosureExSet(event).add(id, closure);
	}

	private void doRemoveObserver(final String event, final UUID id) {
		ClosureExSet set = getClosureExSet(event);

		if (null != set) {
			set.remove(id);
		}
	}

	public Runnable registerObserver(final String event, final ClosureEx closure) {
		FunctorAsync async = new FunctorAsync(closure);
		final UUID id = UUID.randomUUID();
		doRegisterObserver(event, id, async);

		return new Runnable() {

			public void run() {
				// set canceled flag
				closure.setCanceled(true);
				// and remove registered observer
				doRemoveObserver(event, id);

			}
		};
	}

	public Runnable registerObserver(final String event, final Object target,
			final String methodName) {
		return registerObserver(event, new Functor(target, methodName));
	}

	private void doFireEvent(final String event, final Object... args) {
		ClosureExSet set = this.getClosureExSet(event);
		if (null != set) {
			set.execute(args);
		} else {
			if (logger.isDebugEnabled()) {
				logger.debug("event [" + event
						+ "] not found any matched closure!");
			}
		}
	}

	public boolean fireEvent(final String event, final Object... args) {

		logger.debug("Fire event[{}], args[{}].", event,
				ArrayUtils.toString(args));

		// -- 判断当前保存在queue的事件数，如果大于maxEventQueueDepth，则不放入。
		long taskCount = this.getPendingTaskCount();
		if (taskCount >= maxEventQueueDepth) {
			logger.warn(
					"Current event queue size[{}], has been full, event[{}] ignored.",
					taskCount, event);
			return false;
		}

		doFireEvent(event, args);

		return true;
	}

	/**
	 * 还没处理的请求。
	 * 
	 * @return
	 */
	public int getPendingTaskCount() {
		BlockingQueue<Runnable> queue = mainExecutor.getQueue();
		return queue.size();
	}

	/**
	 * 所有的请求。
	 * 
	 * @return
	 */
	public long getTaskCount() {
		return mainExecutor.getTaskCount();
	}

	/**
	 * 活动统计。
	 * 
	 * @return
	 */
	public int getActiveCount() {
		return mainExecutor.getActiveCount();
	}

	/**
	 * 完成的任务统计。
	 * 
	 * @return
	 */
	public long getCompletedTaskCount() {
		return mainExecutor.getCompletedTaskCount();
	}

	public int getPoolSize() {
		return mainExecutor.getPoolSize();
	}

	private Map<String, String> doGetAllEvents() {
		Map<String, String> ret = new HashMap<String, String>();
		for (Map.Entry<String, ClosureExSet> entry : this.closureSet.entrySet()) {
			ret.put(entry.getKey(), entry.getValue().toString());
		}

		return ret;
	}

	public Map<String, String> getAllEvents() throws InterruptedException,
			ExecutionException {
		return doGetAllEvents();
	}

	class FunctorAsync implements ClosureEx {

		private ClosureEx impl;

		public FunctorAsync(ClosureEx impl) {
			this.impl = impl;
		}

		@Override
		public void execute(final Object... args) {
			mainExecutor.execute(new Runnable() {

				public void run() {
					try {
						impl.execute(args);
					} catch (Exception e) {
						logger.error("execute:", e);
					}
				}
			});

		}

		@Override
		public void setCanceled(boolean canceled) {
			this.impl.setCanceled(canceled);
		}

		@Override
		public String toString() {
			return null != impl ? impl.toString() : "functorAsync(null)";
		}
	}
}
